package com.bff.gaia.mix.api.xjoin2.client.mysql.all;

import com.alibaba.fastjson.JSONObject;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.functions.SyncXjoinFunction;
import com.bff.gaia.types.Row;
import com.bff.gaia.util.Collector;

import java.util.List;
import java.util.Map;

public class MysqlAllSyncXjoinFunction implements SyncXjoinFunction<Row> {

	private static final long serialVersionUID = 2537146303821274264L;
	private final MysqlAllXjoinInfo xjoinInfo;

	public MysqlAllSyncXjoinFunction(StreamTableInfo streamTableInfo, SideTableInfo sideTableInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList) {
		this.xjoinInfo = new MysqlAllXjoinInfo(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
	}

	@Override
	public Row xjoin(Row stream, JSONObject side, Collector<Row> out) {
		Row output = new Row(xjoinInfo.getOutFieldSize());
		for (Map.Entry<Integer, Integer> entry : xjoinInfo.getStreamTableOutFieldIndex().entrySet()) {
			output.setField(entry.getKey(), stream.getField(entry.getValue()));
		}
		for (Map.Entry<Integer, String> entry : xjoinInfo.getSideTableOutFieldIndex().entrySet()) {
			output.setField(entry.getKey(), side.getString(entry.getValue()));
		}
		out.collect(output);
		return output;
	}

	public MysqlAllXjoinInfo getXjoinInfo() {
		return xjoinInfo;
	}
}